package com.li.spark.optimization

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

object MapPartitionsOp {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = getSparkContext

    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5))
    //map算子一次处理一条数据
    val sum = dataRDD.map((line) => {
      println(line)
      line * 2
    }).reduce(_ + _);
    println(sum)


    val dataRDD2 = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7))
    //mapPartitions 一次处理一个分区的数据
    val sumPartiotions = dataRDD2.mapPartitions(it => {
      //建议针对初始化连接一类的操作，使用mapPartitions
      var result = new ArrayBuffer[Int]()
      it.foreach(item => {
        result.+=(item * 2)
      })
      result.toIterator
    }).reduce(_ + _)
    println(sumPartiotions)

    sc.stop();
  }

  private def getSparkContext = {
    //创建SparkContext
    val conf = new SparkConf();
    conf.setAppName("MapPartitionsOp").setMaster("local");
    new SparkContext(conf);
  }
}
